package scribe.system.peer.scribe;

import common.peer.PeerAddress;
import java.math.BigInteger;
import java.util.ArrayList;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;

import scribe.simulator.snapshot.Snapshot;
import se.sics.asdistances.ASDistances;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timeout;
import tman.system.peer.tman.RingBuilder;
import tman.system.peer.tman.RingNode;
import tman.system.peer.tman.TManPartnersPort;
import tman.system.peer.tman.TManPartnersRequest;
import tman.system.peer.tman.TManPartnersResponse;

public final class Scribe extends ComponentDefinition {

    Negative<ScribePort> scribePort = negative(ScribePort.class);
    Positive<TManPartnersPort> tmanPartnersPort = positive(TManPartnersPort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
    private long period;
    private PeerAddress self;
    private ArrayList<BigInteger> topics;
    private ArrayList<PeerAddress> tmanPartners;
    private PeerAddress succ;   //added
    private PeerAddress pred;   //added 
    //added
    HashMap<BigInteger, ArrayList<PeerAddress>> topicPublishersMap;
    HashMap<BigInteger, ArrayList<PeerAddress>> topicSubscribersMap;
    HashMap<BigInteger, Boolean> nodeSubscriptions;
    HashMap<BigInteger, Boolean> hopCount;
    //added
    ArrayList<BigInteger> publishersTopicId;
    ArrayList<PeerAddress> changed;
    boolean changedFlag = true;
    //gianmario added
    Map<BigInteger, Set<PeerAddress>> forwardChildren;
    Map<BigInteger, Set<PeerAddress>> rendezVousChildren;
    Set<PublishEvent> topicsToPublish;

//-------------------------------------------------------------------	
    public Scribe() {
        topics = new ArrayList<BigInteger>();
        tmanPartners = new ArrayList<PeerAddress>();

        //added
        topicPublishersMap = new HashMap<BigInteger, ArrayList<PeerAddress>>();
        topicSubscribersMap = new HashMap<BigInteger, ArrayList<PeerAddress>>();
        nodeSubscriptions = new HashMap<BigInteger, Boolean>();
        hopCount = new HashMap<BigInteger, Boolean>();

        //added
        publishersTopicId = new ArrayList<BigInteger>();
        changed = new ArrayList<PeerAddress>();

        subscribe(handleInit, control);
        subscribe(handleRequestTManPartners, timerPort);
        subscribe(handleTimerSubscribe, timerPort);
        subscribe(handlePublish, scribePort);
        subscribe(handleRecvTopicEvent, networkPort);
        subscribe(handleForward, networkPort);
        subscribe(handleRecvTManPartners, tmanPartnersPort);

        subscribe(handleDiscoverRv, networkPort);


    }
//-------------------------------------------------------------------	
    Handler<ScribeInit> handleInit = new Handler<ScribeInit>() {

        public void handle(ScribeInit init) {
            self = init.getSelf();
            topics = init.getTopics();
            period = init.getPeriod();

            //gianmario
            forwardChildren = new HashMap<BigInteger, Set<PeerAddress>>();
            rendezVousChildren = new HashMap<BigInteger, Set<PeerAddress>>();
            topicsToPublish = new HashSet<PublishEvent>();




            //timer for tman partners refresh
            SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
            rst.setTimeoutEvent(new ScribeSchedule(rst));
            trigger(rst, timerPort);

            //timer for push topics published
            ScheduleTimeout t = new ScheduleTimeout(80000L); //???
            Timeout te = new TimerSubscribe(t);
            t.setTimeoutEvent(te);
            trigger(t, timerPort);

            //subscribeAll();

        }
    };
//-------------------------------------------------------------------	
    Handler<PublishEvent> handlePublish = new Handler<PublishEvent>() {

        public void handle(PublishEvent event) {
            BigInteger topicId = event.getTopicId();
            Snapshot.publishTopicEvent(topicId);
            //System.err.println("Peer " + self + " -> TimerPublish: topicId=" + topicId);
            if (rendezVousChildren.containsKey(topicId)) {
                for (PeerAddress child : rendezVousChildren.get(topicId)) {
                    trigger(new TopicEvent(topicId, self, child), networkPort);
                    //System.err.println("Peer " + self + " -> TimerPublish: topicId=" + topicId + " child=" + child);
                }
            }
        }
    };
//-------------------------------------------------------------------	
    Handler<ScribeSchedule> handleRequestTManPartners = new Handler<ScribeSchedule>() {

        public void handle(ScribeSchedule event) {
            TManPartnersRequest request = new TManPartnersRequest();
            trigger(request, tmanPartnersPort);
            //System.err.println("Peer " + self + " -> RequestTmanPartners");

        }
    };
//-------------------------------------------------------------------	
    Handler<TManPartnersResponse> handleRecvTManPartners = new Handler<TManPartnersResponse>() {

        public void handle(TManPartnersResponse event) {
            tmanPartners = event.getPartners();
            Snapshot.updateTManPartners(self, tmanPartners);      //added
            succ = event.getSucc();     //added
            pred = event.getPred();     //added
            Snapshot.updateSuccPred(self, succ, pred);  //added

            //System.err.println("Peer " + self + " -> RecvTmanPartners, tmanPartners"+tmanPartners);


        }
    };
    Handler<ForwardEvent> handleForward = new Handler<ForwardEvent>() {

        @Override
        public void handle(ForwardEvent event) {
            BigInteger topicId = event.getTopicId();
            if (event.getRv() == self) {

                if (!rendezVousChildren.containsKey(topicId)) {
                    rendezVousChildren.put(topicId, new HashSet<PeerAddress>());
                }
                rendezVousChildren.get(event.getTopicId()).add(event.getPeerSource());
                
                Snapshot.updateRendezVous(topicId, self);
                Snapshot.updateChildren(topicId, self, rendezVousChildren.get(topicId));
                //System.err.println("Peer " + self + " -> ForwardEvent, rendezVous for topicId=" + topicId + " , children=" + rendezVousChildren);

            } else {
                try {
                    ArrayList<PeerAddress> view = new ArrayList<PeerAddress>();
                    view.addAll(tmanPartners);
                    view.add(self);
                    view.add(event.getRv());


                    PeerAddress next = nextNodeRv(event.getRv(), view);
                    //System.err.println("Peer " + self + " -> ForwardEvent: topicId=" + topicId + " , view=" + view + " , next=" + next);

                    if (event.getPeerSource() != self) {
                        if (!forwardChildren.containsKey(topicId)) {
                            forwardChildren.put(topicId, new HashSet<PeerAddress>());
                        }


                        forwardChildren.get(event.getTopicId()).add(event.getPeerSource());
                        Snapshot.updateChildren(topicId, self, forwardChildren.get(topicId));
                        
                    }
                    trigger(new ForwardEvent(topicId, self, next, event.getRv()), networkPort);
                    //System.err.println("Peer " + self + " -> ForwardEvent, forwarder for topicId=" + topicId + " , view=" + view + " , tmanPartners=" + tmanPartners + " , next=" + next + " , children=" + forwardChildren);
                } catch (ExceptionNextNodeNotFound ex) {
                    Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
                }

            }


        }
    };
    Handler<TimerSubscribe> handleTimerSubscribe = new Handler<TimerSubscribe>() {

        @Override
        public void handle(TimerSubscribe event) {

            for (BigInteger topicId : topics) {

                try {
                    ArrayList<PeerAddress> view = new ArrayList<PeerAddress>();
                    view.addAll(tmanPartners);
                    view.add(self);
                    PeerAddress next = nextNode(topicId, view);
                    trigger(new DiscoverRv(topicId, self, next, self), networkPort);


                } catch (ExceptionNextNodeNotFound ex) {
                    Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
                }
                //System.err.println("Peer " + self + " -> SubscribeEvent: topicId=" + topicId + " next=" + next);
            }


        }
    };
    Handler<TopicEvent> handleRecvTopicEvent = new Handler<TopicEvent>() {

        @Override
        public void handle(TopicEvent event) {
            BigInteger topicId = event.getTopicId();
            topics.add(topicId);
            Snapshot.updateRecvTopicEvent(self, topicId);
            //System.err.println("Peer " + self + " -> TopicEvent: topicId=" + topicId);
            if (forwardChildren.containsKey(topicId)) {
                for (PeerAddress child : forwardChildren.get(topicId)) {
                    trigger(new TopicEvent(topicId, self, child), networkPort);
                    //System.err.println("Peer " + self + " -> TopicEvent: topicId=" + topicId + " child=" + child);
                }
            }
        }
    };
    Handler<DiscoverRv> handleDiscoverRv = new Handler<DiscoverRv>() {

        @Override
        public void handle(DiscoverRv event) {


            if (event.getPeerSource() == self) {
                try {

                    PeerAddress rv = event.getRv();
                    //System.err.println("Self " + self + " --> RV found, topic=" + event.getTopicId() + " , rv=" + rv);

                    ArrayList<PeerAddress> view = new ArrayList<PeerAddress>();
                    BigInteger topicId = event.getTopicId();
                    view.addAll(tmanPartners);
                    view.add(self);
                    PeerAddress next = nextNodeRv(rv, view);
                    trigger(new ForwardEvent(topicId, self, next, rv), networkPort);


                } catch (ExceptionNextNodeNotFound ex) {
                    Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
                }
            } else {

                try {
                    ArrayList<PeerAddress> view = new ArrayList<PeerAddress>();
                    view.addAll(tmanPartners);
                    view.add(self);
                    BigInteger topicId = event.getTopicId();
                    PeerAddress next = nextNode(event.getTopicId(), view);

                    if (next == self) {
                        trigger(new DiscoverRv(topicId, event.getPeerSource(), event.getPeerSource(), self), networkPort);
                    } else {
                        trigger(new DiscoverRv(topicId, event.getPeerSource(), next, null), networkPort);


                    }

                } catch (ExceptionNextNodeNotFound ex) {
                    Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    };

    private PeerAddress nextNode(BigInteger topicId, Collection<PeerAddress> view) throws ExceptionNextNodeNotFound {


        PeerAddress topicVirtualAddress = new PeerAddress(null, topicId);
        PeerAddress next = null;

        try {
            RingNode predTopicAdd = RingBuilder.succPred(topicVirtualAddress, view)[0];
            next = predTopicAdd.getSelf();
        } catch (Exception ex) {
            throw new ExceptionNextNodeNotFound();
            //Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
        }
        if (next == null) {
            throw new ExceptionNextNodeNotFound();
        }
        //System.err.println("Peer " + self + " --> nextNode: topicId=" + topicId + " , view=" + view + " , next=" + next);

        return next;
    }

    private PeerAddress nextNodeRv(PeerAddress rv, Collection<PeerAddress> view) throws ExceptionNextNodeNotFound {
        ASDistances distances = ASDistances.getInstance();

        int currentAS = distances.getASFromIP(self.getPeerAddress().getIp().getHostAddress());
        int rvAS = distances.getASFromIP(rv.getPeerAddress().getIp().getHostAddress());
        //System.err.println("Self " + self + " --> " + "computing nextNode based on Rv " + rv + " rvAs=" + rvAS);

        PeerAddress next = null;
        //System.err.println("Self " + self + " --> " + "view " + view);

        List<PeerAddress> sortedViewAS = new ArrayList<PeerAddress>(view);
        Collections.sort(sortedViewAS, new DistanceASComparator(rv));
        //System.err.println("Self " + self + " --> " + "sortedViewAS " + sortedViewAS);

        int nextAS = distances.getASFromIP(sortedViewAS.get(0).getPeerAddress().getIp().getHostAddress());
        //System.err.println("Self " + self + " --> " + "nextAS=" + nextAS);

        Collection<PeerAddress> viewFilteredAS = filterViewInAS(view, nextAS);
        //System.err.println("Self " + self + " --> " + "filterdViewAS " + viewFilteredAS);

        try {
            RingNode predTopicAdd = RingBuilder.succPred(rv, viewFilteredAS)[0];
            next = predTopicAdd.getSelf();
        } catch (Exception ex) {
            throw new ExceptionNextNodeNotFound();
            //Logger.getLogger(Scribe.class.getName()).log(Level.SEVERE, null, ex);
        }
        if (next == null) {
            throw new ExceptionNextNodeNotFound();
        }
        //System.err.println("Peer " + self + " --> nextNode: topicId=" + topicId + " , view=" + view + " , next=" + next);

        return next;
    }

    private Collection<PeerAddress> filterViewInAS(Collection<PeerAddress> view, int as) {
        Collection<PeerAddress> asView = new HashSet<PeerAddress>();
        ASDistances distances = ASDistances.getInstance();
        for (PeerAddress peer : view) {
            if (distances.getASFromIP(peer.getPeerAddress().getIp().getHostAddress()) == as) {
                asView.add(peer);
            }
        }
        return asView;
    }
    //-------------------------------------------------------------------

    private BigInteger computeTopicIdHash(BigInteger topicId) {
        final int prime = 31;
        int result = 1;

        result = prime * result + ((topicId == null) ? 0 : topicId.hashCode());
        return BigInteger.valueOf(result);
    }
}
